#ifndef PREPROCESSDATAGROUPSBOX_H_JM_20130316
#define PREPROCESSDATAGROUPSBOX_H_JM_20130316

#include "../BufferedOutputBox.h"
#include "../../LogConf.h"

namespace jm4bbx {

	class DataDistributor {
	public:
		const static int MASK_CLEANER = 31;

		DataDistributor(int startMask, int endMask, int oId)
			: startMask_(startMask), 
			endMask_(endMask), 
			outId_(oId) { }

		inline bool isAccpeted(int value) const {
			for (int mask = startMask_; mask < endMask_; ++mask) {
				if (!((value & MASK_CLEANER) ^ mask)) {
					return true;
				}
			}
			return false;
		}

		int getOutputId() const { return outId_; }
		virtual ~DataDistributor() { }
	private:
		int startMask_;
		int endMask_;
		int outId_;
	};


	
	template<typename valueT>
	class  PreprocessDataGroupsBox : public BufferedOutputBox<KeyRecord<valueT>> {
		JM_USE_LOGGER(PreprocessDataGroupsBoxLogger)
	public:
		//typedef KeyMultiRecords<T> ValueType;
		//typedef KeyRecord<valueT> TransferValueType;

		static const  groupType INPUT_GROUP;
		static const  groupType OUTPUT_GROUP_BEGIN;	
		static const  columnType INPUT_COLUMN;
		static const  columnType OUTPUT_COLUMN_BEGIN;
		static const  inType IN_ARC;
		static const  outType OUT_ARC_BEGIN;

		static const char* DATA_GROUP_COUNT_PARAM_NAME;

		PreprocessDataGroupsBox(const BoxParamPack& p) 
			: BufferedOutputBox(p) {
				
				dataGroupCount_ = 
					getBoxParameterValue(p, DATA_GROUP_COUNT_PARAM_NAME, 1);
		}		

		virtual void initImpl() override {

			add_group(INPUT_GROUP, INPUT_COLUMN, 
				get_input_descriptors(IN_ARC));

			// +1 pretoze sa indexuje od nuly a MASK_CLEANER je maximalna hodnota masky
			int maskSize = (DataDistributor::MASK_CLEANER + 1)/dataGroupCount_;

			//size_t buffSize = jm4bbx::BaseBufferedOutputBox::DEFAULT_CHUNK_SIZE/dataGroupCount_;

			for (int i = 0; i < dataGroupCount_; ++i) {
				groupType g = getOutGroup(i);
				columnType c = getOutCol(i);
				outType a = getOutArc(i);
				add_group(g, c, get_output_descriptors(a));
				OutId id = addBufferedOutput(g, c, a/*, buffSize*/);
				// Delenie na intervaly ktore su spracovavne ako 
				// polozuavrete intervaly <x,y) pre poslednu 
				// hodnotu je potrebne ale spracovat <x, y> preto sa 
				// zadava <x, y+1)
				int startMask = i*maskSize;
				int endMask = 
					((i+1)==dataGroupCount_) ? 
						DataDistributor::MASK_CLEANER + 1: startMask + maskSize;

				outs_.push_back(DataDistributor(startMask, endMask, id));
			}

			initBufferedOutputs();
			wait4SignalsIn();
			receive(IN_ARC, INPUT_GROUP);	
		}

		virtual void syncMachEtwas() override {
			LOG_D("Do buckets.")
			if (!ig_) ig_ = getGroupPtr(INPUT_GROUP);
			int counter = 0;
			for (; ig_->getReadPosition() < ig_->getReadEnd(); ig_->moveReadPosition()) {
				++counter;
				ValueType value = 
					get_data<ValueType>(INPUT_COLUMN)[ig_->getReadPosition()];		

				auto key = value.getKey();
				// len data pre tento box
				for (auto it = outs_.begin(); it != outs_.end(); ++it) {
					if (it->isAccpeted(key)) {
						addOutputData(it->getOutputId(), value);
					}
				}
			}		
			LOG_D("Done " << counter << " buckets -> sending data.")
			bool sent = sendAllOutputData();
			bool empty = areAllOutputsEmpty();
			LOG_D("Sent > " << sent << " -> buff is empty > " << empty);
			if (empty) {
				LOG_D("Buffer empty -> send signals");
				sendSignalsOut();
				for (int i = 0; i < dataGroupCount_; ++i) {
					send_poisoned(getOutArc(i)); // Aby ciel vedel ze je hotovo!!!
				}
				receive(IN_ARC, INPUT_GROUP);	
			}
		}

	protected:
		static outType getOutArc(int index) { 
			return outType(OUT_ARC_BEGIN + index); 
		} 
		static groupType getOutGroup(int index) { 
			return groupType(OUTPUT_GROUP_BEGIN + index); 
		}
		static columnType getOutCol(int index) { 
			return columnType(OUTPUT_COLUMN_BEGIN + index); 
		}
	private:
		int dataGroupCount_;
		std::vector<DataDistributor> outs_;
		GroupPtr ig_;
	};

	template<typename valueT>
	const char* PreprocessDataGroupsBox<valueT>::DATA_GROUP_COUNT_PARAM_NAME = "data_group_count";

	template<typename valueT>
	const  outType PreprocessDataGroupsBox<valueT>::OUT_ARC_BEGIN = outType(0);

	template<typename valueT>
	const  inType PreprocessDataGroupsBox<valueT>::IN_ARC = inType(0);

	template<typename valueT>
	const  columnType PreprocessDataGroupsBox<valueT>::OUTPUT_COLUMN_BEGIN = columnType(1);

	template<typename valueT>
	const  columnType PreprocessDataGroupsBox<valueT>::INPUT_COLUMN = columnType(0);

	template<typename valueT>
	const  groupType PreprocessDataGroupsBox<valueT>::OUTPUT_GROUP_BEGIN = groupType(1);

	template<typename valueT>
	const  groupType PreprocessDataGroupsBox<valueT>::INPUT_GROUP = groupType(0);

} // namespace jm4bbx


#endif
